package com.taobao.analyse.main;

import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.SequenceFileInputFormat;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapred.lib.MultipleInputs;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.taobao.analyse.map.AuctionCategoryMapper;
import com.taobao.analyse.map.AuctionCategoryTitleMapper;
import com.taobao.analyse.map.CategoryPriceChangeRateMapper;
import com.taobao.analyse.map.ItemPriceChangedMapper;
import com.taobao.analyse.map.MergeTimeResultMapper;
import com.taobao.analyse.map.MergeTimeResultTempMapper;
import com.taobao.analyse.reduce.AuctionCategoryReducer;
import com.taobao.analyse.reduce.AuctionCategoryTitleReducer;
import com.taobao.analyse.reduce.CategoryPriceChangeRateReducer;
import com.taobao.analyse.reduce.ItemPriceChangedReducer;
import com.taobao.analyse.reduce.MergeResultReducer;
import com.taobao.analyse.reduce.MergeTimeResultReducer;
import com.taobao.cmp.lib.sst.mapred.input.SSTFileInputFormat;

/**
 * 主流程函数，包含几个子Job hadoop jar *.jar
 * com.taobao.analyse.main.ItemPriceChangeAnalyseMain 1223 zhuliu
 * 
 * @author Livis
 * 
 */
public class ItemPriceChangeAnalyseMain extends Configured implements Tool {

	//private static final String basepath = "/group/tbauction/users/qianyi.zby/tieyi.qlr/output/";
	private static String basepath = "/group/tbauction/zhuliu/tieyi/output/";

	@Override
	public int run(String[] args) throws Exception {

		if (args.length < 2) {
			System.out
					.println("please input the time and output path---tieyi.qlr!!!");
			return -1;
		}
		
		if(args[1].equals("zhuliu")){
			basepath = "/group/tbauction/zhuliu/tieyi/output/";
		}else if(args[1].equals("tieyi")){
			basepath = "/group/tbauction/tieyi.qlr/output/";
		}else{
			basepath = "/group/tbauction/users/qianyi.zby/tieyi.qlr/output/";
		}

		String outputPath0 = basepath + args[0] + "/categoryname/";
		String outputPath1 = basepath + args[0] + "/itemupdate/";
		String outputPath2 = basepath + args[0] + "/category/";
		String outputPath3 = basepath + args[0] + "/firstmerge/";
		String outputPath4 = basepath + args[0] + "/result/";
		String outputPath5 = basepath + args[0] + "/timeresult/";
		String outputPath6 = basepath + args[0] + "/finaltimeresult/";
		// 执行第一个任务，扫描生命周期的编辑记录
		if (true) {
			JobConf conf = new JobConf(this.getConf(), this.getClass());
			// MISC
			conf.setJobName("DupPriceChange");
			conf.setMaxMapAttempts(2);
			conf.setNumReduceTasks(200);

			// MAPPER
			conf.setMapOutputKeyClass(LongWritable.class);
			conf.setMapOutputValueClass(Text.class);

			// REDUCER
			conf.setOutputKeyClass(LongWritable.class);
			conf.setOutputValueClass(Text.class);
			// 设置reducer
			conf.setReducerClass(ItemPriceChangedReducer.class);

			List<String> paths = new ArrayList<String>();
			paths.add("/group/tbauction/lifecycle/archive/20120502-14");
			paths.add("/group/tbauction/lifecycle/archive/20120518-14");
			//paths.add("/group/tbauction/lifecycle/archive/20120601-05");

			// 加入SSTFile路径列表
			for (String x : paths) {
				System.out.println("add SSTFile-Input" + x);
				// OUTPUT
				MultipleInputs.addInputPath(conf, new Path(x),
						SSTFileInputFormat.class, ItemPriceChangedMapper.class);
			}

			// OUTPUT
			Path out = new Path(outputPath1);
			conf.setOutputFormat(TextOutputFormat.class);

			System.out.println("set the output path:" + out.toString());
			TextOutputFormat.setOutputPath(conf, out);

			FileSystem fs = FileSystem.get(conf);
			if (!fs.exists(out)) {
				JobClient.runJob(conf);
			} else {
				System.out
						.println("get item update record had finished!-tieyi.qlr");
			}
		}
		
		// 获取商品的一级类目信息
		if (true) {
			JobConf conf = new JobConf(this.getConf(), this.getClass());
			// MISC
			conf.setJobName("DupAuctionCategory");
			conf.setMaxMapAttempts(2);
			conf.setNumReduceTasks(200);

			// MAPPER
			conf.setMapOutputKeyClass(LongWritable.class);
			conf.setMapOutputValueClass(Text.class);

			// REDUCER
			conf.setOutputKeyClass(LongWritable.class);
			conf.setOutputValueClass(Text.class);
			// 设置reducer
			conf.setReducerClass(AuctionCategoryReducer.class);

			List<String> paths = new ArrayList<String>();
			// paths.add("/group/taobao/taobao/hive/r_auction_auctions/pt=20120101000000/if_online=1");
			// paths.add("/group/taobao/taobao/hive/r_auction_auctions/pt=20120201000000/if_online=1");
			// paths.add("/group/taobao/taobao/hive/r_auction_auctions/pt=20120301000000/if_online=1");
			
			paths.add("/group/taobao/taobao/hive/r_auction_auctions/pt=20120505000000/if_online=1");
			paths.add("/group/taobao/taobao/hive/r_auction_auctions/pt=20120517000000/if_online=1");
			paths.add("/group/taobao/taobao/hive/r_auction_auctions/pt=20120604000000/if_online=1");

			// 加入SSTFile路径列表
			for (String x : paths) {
				System.out.println("add File-Input" + x);
				// OUTPUT
				MultipleInputs.addInputPath(conf, new Path(x),
						SequenceFileInputFormat.class,
						AuctionCategoryMapper.class);
			}

			// OUTPUT
			Path out = new Path(outputPath2);
			conf.setOutputFormat(TextOutputFormat.class);

			System.out.println("set the output path:" + out.toString());
			TextOutputFormat.setOutputPath(conf, out);

			FileSystem fs = FileSystem.get(conf);
			if (!fs.exists(out)) {
				JobClient.runJob(conf);
			} else {
				System.out
						.println("get item category info had finished!-tieyi.qlr");
			}
		}
		
		// 获取一级类目的标题名
		if (true) {
			JobConf conf = new JobConf(this.getConf(), this.getClass());
			// MISC
			conf.setJobName("getAuctionCategoryName");
			conf.setMaxMapAttempts(2);
			conf.setNumReduceTasks(1);

			// MAPPER
			conf.setMapOutputKeyClass(LongWritable.class);
			conf.setMapOutputValueClass(Text.class);

			// REDUCER
			conf.setOutputKeyClass(LongWritable.class);
			conf.setOutputValueClass(Text.class);
			// 设置reducer
			conf.setReducerClass(AuctionCategoryTitleReducer.class);

			List<String> paths = new ArrayList<String>();
			paths.add("/group/taobao/taobao/hive/r_auction_category/pt=20120604000000");

			// 加入SSTFile路径列表
			for (String x : paths) {
				System.out.println("add File-Input" + x);
				// OUTPUT
				MultipleInputs.addInputPath(conf, new Path(x),
						SequenceFileInputFormat.class,
						AuctionCategoryTitleMapper.class);
			}

			// OUTPUT
			Path out = new Path(outputPath0);
			conf.setOutputFormat(TextOutputFormat.class);

			System.out.println("set the output path:" + out.toString());
			TextOutputFormat.setOutputPath(conf, out);

			FileSystem fs = FileSystem.get(conf);
			if (!fs.exists(out)) {
				JobClient.runJob(conf);
			} else {
				System.out
						.println("get item category title from auction_category had finished!-tieyi.qlr");
			}
		}

		// 联立上面两个输出文件，获得商品对应的一级类目信息和商品编辑记录
		if (true) {
			JobConf conf = new JobConf(this.getConf(), this.getClass());
			// MISC
			conf.setJobName("DupPriceChangeResult");
			conf.setMaxMapAttempts(2);
			conf.setNumReduceTasks(50);

			// MAPPER
			conf.setMapOutputKeyClass(LongWritable.class);
			conf.setMapOutputValueClass(Text.class);

			// REDUCER
			conf.setOutputKeyClass(LongWritable.class);
			conf.setOutputValueClass(Text.class);
			// 设置reducer
			conf.setReducerClass(CategoryPriceChangeRateReducer.class);

			List<String> paths = new ArrayList<String>();
			paths.add(outputPath1);
			paths.add(outputPath2);

			// 加入SSTFile路径列表
			for (String x : paths) {
				System.out.println("add File-Input" + x);
				// OUTPUT
				MultipleInputs.addInputPath(conf, new Path(x),
						TextInputFormat.class,
						CategoryPriceChangeRateMapper.class);
			}

			// OUTPUT
			Path out = new Path(outputPath3);
			conf.setOutputFormat(TextOutputFormat.class);

			System.out.println("set the output path:" + out.toString());
			TextOutputFormat.setOutputPath(conf, out);

			FileSystem fs = FileSystem.get(conf);
			if (!fs.exists(out)) {
				JobClient.runJob(conf);
			} else {
				System.out
						.println("merge item category and update info had finished!-tieyi.qlr");
			}
		}

		// merge上一次计算的结果
		if (true) {
			JobConf conf = new JobConf(this.getConf(), this.getClass());
			// MISC
			conf.setJobName("mergeResult");
			conf.setMaxMapAttempts(2);
			conf.setNumReduceTasks(1);

			// MAPPER
			conf.setMapOutputKeyClass(LongWritable.class);
			conf.setMapOutputValueClass(Text.class);

			// REDUCER
			conf.setOutputKeyClass(Text.class);
			conf.setOutputValueClass(Text.class);
			// 设置reducer
			conf.setReducerClass(MergeResultReducer.class);

			List<String> paths = new ArrayList<String>();
			paths.add(outputPath0);
			paths.add(outputPath3);

			// 加入File路径列表
			for (String x : paths) {
				System.out.println("add File-Input" + x);
				// OUTPUT
				MultipleInputs.addInputPath(conf, new Path(x),
						TextInputFormat.class,
						CategoryPriceChangeRateMapper.class);
			}

			// OUTPUT
			Path out = new Path(outputPath4);
			conf.setOutputFormat(TextOutputFormat.class);

			System.out.println("set the output path:" + out.toString());
			TextOutputFormat.setOutputPath(conf, out);

			FileSystem fs = FileSystem.get(conf);
			if (!fs.exists(out)) {
				JobClient.runJob(conf);
			} else {
				System.out
						.println("merge item update result had finished!-tieyi.qlr");
			}
		}

		// mergeTimte上一次计算的结果
		if (true) {
			JobConf conf = new JobConf(this.getConf(), this.getClass());
			// MISC
			conf.setJobName("mergeTimeResult");
			conf.setMaxMapAttempts(2);
			conf.setNumReduceTasks(20);

			// MAPPER
			conf.setMapOutputKeyClass(Text.class);
			conf.setMapOutputValueClass(LongWritable.class);

			// REDUCER
			conf.setOutputKeyClass(Text.class);
			conf.setOutputValueClass(LongWritable.class);
			// 设置reducer
			conf.setReducerClass(MergeTimeResultReducer.class);

			List<String> paths = new ArrayList<String>();
			paths.add(outputPath0);
			paths.add(outputPath3);

			// 加入File路径列表
			for (String x : paths) {
				System.out.println("add File-Input" + x);
				// OUTPUT
				MultipleInputs.addInputPath(conf, new Path(x),
						TextInputFormat.class,
						MergeTimeResultMapper.class);
			}

			// OUTPUT
			Path out = new Path(outputPath5);
			conf.setOutputFormat(TextOutputFormat.class);

			System.out.println("set the output path:" + out.toString());
			TextOutputFormat.setOutputPath(conf, out);

			FileSystem fs = FileSystem.get(conf);
			if (!fs.exists(out)) {
				JobClient.runJob(conf);
			} else {
				System.out
						.println("merge item time result had finished!-tieyi.qlr");
			}
		}

		// mergeTimteFinal上一次计算的结果
		if (true) {
			JobConf conf = new JobConf(this.getConf(), this.getClass());
			// MISC
			conf.setJobName("mergeFinalTimeResult");
			conf.setMaxMapAttempts(2);
			conf.setNumReduceTasks(1);

			// MAPPER
			conf.setMapOutputKeyClass(Text.class);
			conf.setMapOutputValueClass(LongWritable.class);

			// REDUCER
			conf.setOutputKeyClass(Text.class);
			conf.setOutputValueClass(LongWritable.class);
			// 设置reducer
			conf.setReducerClass(MergeTimeResultReducer.class);

			List<String> paths = new ArrayList<String>();
			paths.add(outputPath5);

			// 加入File路径列表
			for (String x : paths) {
				System.out.println("add File-Input" + x);
				// OUTPUT
				MultipleInputs.addInputPath(conf, new Path(x),
						TextInputFormat.class, MergeTimeResultTempMapper.class);
			}

			// OUTPUT
			Path out = new Path(outputPath6);
			conf.setOutputFormat(TextOutputFormat.class);

			System.out.println("set the output path:" + out.toString());
			TextOutputFormat.setOutputPath(conf, out);

			FileSystem fs = FileSystem.get(conf);
			if (!fs.exists(out)) {
				JobClient.runJob(conf);
			} else {
				System.out
						.println("merge item time result had finished!-tieyi.qlr");
			}
		}

		return JobClient.SUCCESS;
	}
	
	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(),
				new ItemPriceChangeAnalyseMain(), args);
		System.exit(res);
	}

}
